package com.bigdata.core.transformations

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable.ArrayBuffer

/**
 * coalesce(numPartiton,shuffle=false)
 * coalesce 增加或者减少分区，默认没有shuffle
 * coalesce 如果从少的分区增到多的分区，如果指定没有shuffle ,那么不起作用，分区数不变。
 * coalesce(num,true) = repartition(num)
 * 用于减少下游分区, 不会shuffle,下游的一个分区处理上游的多个分区数据
 * 窄依赖，多对一关系，结合mapPartitionWithIndex算子使用，观察分区内数据分配结果
 */
object Demo20_coalesce {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
    conf.setMaster("local").setAppName("coalesce")
    val sc = new SparkContext(conf)
    sc.setLogLevel("error")

    val rdd1: RDD[String] = sc.parallelize(List[String](
      "love1", "love2", "love3", "love4",
      "love5", "love6", "love7", "love8",
      "love9", "love10", "love11", "love12"), 4)

    val rdd2: RDD[String] = rdd1.mapPartitionsWithIndex((index, iter) => {
      val buffer = new ArrayBuffer[String]()
      iter.foreach(str => {
        buffer.append(s"分区号${index}对应的值${str}")
      })
      buffer.iterator
    })

    rdd2.foreach(println)

    // shuffle 默认是false， 没有shuffle，可能有数据倾斜
    // 如果下游的RDD分区数要增大，就必要要走shuffle，否则没有意义
    // coalesce默认是不走shuffle，所以大多数场景下使用来做减少分区处理
    //    val coalesceRdd: RDD[String] = rdd1.coalesce(6, true)
    val coalesceRdd: RDD[String] = rdd2.repartition(6)
    println(coalesceRdd.getNumPartitions)
    // coalesceRdd.foreach(println)

    println("-----------------------")

    val result: RDD[String] = coalesceRdd.mapPartitionsWithIndex((index, iter) => {
      val array: ArrayBuffer[String] = ArrayBuffer[String]()
      iter.foreach(
        elem => {
          array.append(s"partition = 【$index】,value = " + elem)
        }
      )
      array.iterator
    })

    result.foreach(println)
  }
}
